package main.scala.org.huel.dataprocessing

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

import scala.collection.mutable
import java.sql.{Connection, DriverManager, PreparedStatement}
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import scala.math.{sqrt, pow}

/**
 * @author Liweijian
 * @Description 基于协同过滤的课程推荐系统
 * @date 2025/06/13
 */
object CourseRecommendationSystem {

  // 数据模型
  case class AttendanceRecord(
                               classId: String,
                               studentName: String,
                               courseName: String,
                               studentId: String,
                               score: Int,
                               isAbsent: Int
                             )

  case class UserCourseRating(
                               studentId: String,
                               courseName: String,
                               rating: Double // 基于出勤率计算的评分
                             )

  case class CourseRecommendation(
                                   studentId: String,
                                   courseName: String,
                                   score: Double,
                                   rank: Int
                                 )

  // 推荐参数配置
  val NUM_RECOMMENDATIONS = 5    // 推荐数量
  val SIMILARITY_THRESHOLD = 0.3 // 相似度阈值
  val SIMILAR_USER_COUNT = 10    // 寻找的相似用户数量

  def main(args: Array[String]): Unit = {
    // 创建SparkConf
    val sparkConf = new SparkConf()
      .setAppName("CourseRecommendationSystem")
      .setMaster("local[*]")  // 本地运行模式

    // 创建SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    import spark.implicits._

    // 设置日志级别
    spark.sparkContext.setLogLevel("WARN")

    // 创建StreamingContext，批处理时间为30秒
    val ssc = new StreamingContext(spark.sparkContext, Seconds(30))

    // Kafka参数配置
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "43.143.125.94:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "course_recommendation_group123",
      "auto.offset.reset" -> "earliest", // 从最早的数据开始，确保有足够数据用于推荐
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    // 订阅Kafka主题
    val topics = Array("attendance")

    // 打印启动信息
    println(s"\n✓ 正在监听Kafka主题: ${topics.mkString(", ")}")
    println(s"✓ 使用Kafka服务器: 43.143.125.94:9092")
    println(s"✓ 消费者组ID: course_recommendation_group")

    // 创建直接流
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    // 定义处理逻辑
    stream.foreachRDD { rdd =>
      // 获取偏移量范围以便稍后提交
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      println(s"收到RDD，包含 ${rdd.count()} 条记录")

      if (!rdd.isEmpty()) {
        // 获取当前时间
        val currentTime = LocalDateTime.now()
        val formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
        val formattedTime = currentTime.format(formatter)

        println(s"\n=========================================================")
        println(s"| 协同过滤课程推荐系统 - ${formattedTime} |")
        println(s"=========================================================")

        // 1. 解析考勤数据
        val attendanceRecordsRDD = rdd.map(record => record.value())
          .map { line =>
            val fields = line.split("\t")
            if (fields.length == 6) {
              AttendanceRecord(
                classId = fields(0),
                studentName = fields(1),
                courseName = fields(2),
                studentId = fields(3),
                score = fields(4).toInt,
                isAbsent = fields(5).toInt
              )
            } else {
              // 处理格式不正确的数据
              AttendanceRecord("Unknown", "Unknown", "Unknown", "Unknown", -1, -1)
            }
          }
          .filter(record => record.score >= 0) // 过滤掉无效记录

        // 2. 计算用户对每个课程的评分（基于出勤率）
        val userCourseRatings = calculateUserCourseRatings(attendanceRecordsRDD)

        // 3. 构建用户-课程评分矩阵
        val userCourseMatrix = buildUserCourseMatrix(userCourseRatings)
        println("\n【用户-课程关系矩阵】")
        printUserCourseMatrix(userCourseMatrix)

        // 4. 构建课程-课程相似度矩阵
        val courseSimilarityMatrix = buildCourseSimilarityMatrix(userCourseMatrix)
        println("\n【课程-课程相似度矩阵】")
        printCourseSimilarityMatrix(courseSimilarityMatrix)

        // 5. 生成用户推荐列表
        val recommendations = generateRecommendations(userCourseMatrix, courseSimilarityMatrix)
        println("\n【用户推荐列表】")
        printRecommendations(recommendations)

        // 6. 保存推荐结果到MySQL
        try {
          // 保存用户课程推荐结果
          saveRecommendationsToDb(recommendations)

          // 保存课程相似度矩阵
          saveCourseSimilarityToDb(courseSimilarityMatrix)

          // 保存用户-课程评分矩阵
          saveUserCourseMatrixToDb(userCourseMatrix)

          println("\n✓ 推荐结果已保存到MySQL数据库")
        } catch {
          case e: Exception =>
            println(s"\n✗ 保存推荐结果到MySQL出错: ${e.getMessage}")
            e.printStackTrace()
        }

        // 手动提交偏移量
        stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
      }
    }

    // 启动Streaming Context
    ssc.start()
    println("\n✓ 协同过滤课程推荐系统已启动")
    println("✓ 正在监听Kafka数据流...")
    ssc.awaitTermination()
  }

  /**
   * 计算用户对每个课程的评分（基于出勤率）
   */
  def calculateUserCourseRatings(attendanceRDD: RDD[AttendanceRecord]): RDD[UserCourseRating] = {
    attendanceRDD
      .map(record => ((record.studentId, record.courseName), (record.isAbsent, 1)))
      .reduceByKey((pair1, pair2) => (pair1._1 + pair2._1, pair1._2 + pair2._2))
      .map { case ((studentId, courseName), (totalAbsent, totalCount)) =>
        // 计算出勤率作为评分，缩放到1-5的评分区间，出勤率越高评分越高
        val attendanceRate = if (totalCount > 0) (totalCount - totalAbsent).toDouble / totalCount else 0
        val rating = 1 + attendanceRate * 4 // 范围映射到1-5
        UserCourseRating(studentId, courseName, rating)
      }
  }

  /**
   * 构建用户-课程评分矩阵
   */
  def buildUserCourseMatrix(ratingsRDD: RDD[UserCourseRating]): Map[String, Map[String, Double]] = {
    // 使用RDD操作构建嵌套Map表示用户-课程矩阵
    val userCourseMap = ratingsRDD
      .map(rating => (rating.studentId, (rating.courseName, rating.rating)))
      .groupByKey()
      .mapValues(courseRatings => courseRatings.toMap)
      .collect()
      .toMap

    userCourseMap
  }

  /**
   * 构建课程-课程相似度矩阵 (使用余弦相似度)
   */
  def buildCourseSimilarityMatrix(userCourseMatrix: Map[String, Map[String, Double]]): Map[String, Map[String, Double]] = {
    // 1. 收集所有课程
    val allCourses = userCourseMatrix.values.flatMap(_.keys).toSet.toSeq

    // 2. 构建课程向量 (课程为key，用户评分为vector)
    val courseVectors = mutable.Map[String, Map[String, Double]]()

    // 对每门课程
    for (course <- allCourses) {
      // 收集所有对该课程有评分的用户和评分值
      val userRatings = mutable.Map[String, Double]()

      // 遍历每个用户
      for ((userId, courseRatings) <- userCourseMatrix) {
        // 如果用户对该课程有评分，添加到向量中
        if (courseRatings.contains(course)) {
          userRatings(userId) = courseRatings(course)
        }
      }

      courseVectors(course) = userRatings.toMap
    }

    // 3. 计算课程-课程相似度
    val courseSimilarities = mutable.Map[String, Map[String, Double]]()

    for (course1 <- allCourses) {
      val similarities = mutable.Map[String, Double]()

      for (course2 <- allCourses) {
        if (course1 != course2) {
          // 计算course1和course2的余弦相似度
          val similarity = calculateCosineSimilarity(
            courseVectors(course1),
            courseVectors(course2)
          )

          // 仅保留超过阈值的相似度
          if (similarity > SIMILARITY_THRESHOLD) {
            similarities(course2) = similarity
          }
        }
      }

      courseSimilarities(course1) = similarities.toMap
    }

    courseSimilarities.toMap
  }

  /**
   * 计算两个向量之间的余弦相似度
   */
  def calculateCosineSimilarity(vector1: Map[String, Double], vector2: Map[String, Double]): Double = {
    // 找到两个向量共有的键
    val commonKeys = vector1.keySet.intersect(vector2.keySet)

    if (commonKeys.isEmpty) {
      return 0.0 // 无共同用户，相似度为0
    }

    // 计算余弦相似度的分子 (点积)
    val dotProduct = commonKeys.map(key => vector1(key) * vector2(key)).sum

    // 计算向量长度 (分母)
    val vector1Norm = sqrt(vector1.values.map(x => pow(x, 2)).sum)
    val vector2Norm = sqrt(vector2.values.map(x => pow(x, 2)).sum)

    // 计算相似度
    if (vector1Norm * vector2Norm == 0) 0.0 else dotProduct / (vector1Norm * vector2Norm)
  }

  /**
   * 基于User-Based CF生成推荐
   */
  def generateRecommendations(
                               userCourseMatrix: Map[String, Map[String, Double]],
                               courseSimilarityMatrix: Map[String, Map[String, Double]]
                             ): List[(String, List[(String, Double, Int)])] = {

    val userRecommendations = mutable.ListBuffer[(String, List[(String, Double, Int)])]()

    // 获取所有可用课程
    val allCourses = courseSimilarityMatrix.keys.toSet
    println(s"\n系统中总共有 ${allCourses.size} 门课程: ${allCourses.mkString(", ")}")

    // 为每个用户生成推荐
    var usersWithNoRecommendations = 0
    var usersWithAllCourses = 0

    for ((userId, courseRatings) <- userCourseMatrix) {
      // 已有的课程
      val userCourses = courseRatings.keySet

      // 用户未参加的课程
      val untakenCourses = allCourses.diff(userCourses)

      if (untakenCourses.isEmpty) {
        usersWithAllCourses += 1

        // 方案1: 当用户已参加所有课程时，仍然根据评分和相似度推荐最适合复习的课程
        // 使用用户当前评分最低的课程作为推荐基础
        val lowestRatedCourses = courseRatings.toSeq.sortBy(_._2).take(2).map(_._1)
        val candidateCourses = mutable.Map[String, Double]()

        for (course <- lowestRatedCourses) {
          val similarCourses = courseSimilarityMatrix.getOrElse(course, Map())
          for ((similarCourse, similarity) <- similarCourses) {
            // 排除自己
            if (similarCourse != course) {
              val currentScore = candidateCourses.getOrElse(similarCourse, 0.0)
              // 计算"回顾建议"分数
              val reviewScore = similarity * (5.0 - courseRatings(course)) // 评分低的课程获得更高的回顾分数
              candidateCourses(similarCourse) = currentScore + reviewScore
            }
          }
        }

        // 添加"回顾"标记和排名
        val recommendations = candidateCourses.toList
          .sortBy(-_._2)
          .take(NUM_RECOMMENDATIONS)
          .zipWithIndex
          .map { case ((course, score), index) => (s"${course} (推荐复习)", score, index + 1) }

        if (recommendations.nonEmpty) {
          userRecommendations += ((userId, recommendations))
        }

      } else {
        // 常规推荐逻辑 - 推荐未参加的课程
        val candidateCourses = mutable.Map[String, Double]()

        // 对用户的每门课
        for ((userCourse, rating) <- courseRatings) {
          // 获取相似课程
          val similarCourses = courseSimilarityMatrix.getOrElse(userCourse, Map())

          // 对每门相似课程
          for ((similarCourse, similarity) <- similarCourses) {
            // 如果是用户没有的课程
            if (!userCourses.contains(similarCourse)) {
              // 累加推荐得分 (用户评分 × 课程相似度)
              val currentScore = candidateCourses.getOrElse(similarCourse, 0.0)
              candidateCourses(similarCourse) = currentScore + rating * similarity
            }
          }
        }

        // 获取评分最高的N门课作为推荐，并添加排名
        val recommendations = candidateCourses.toList
          .sortBy(-_._2)
          .take(NUM_RECOMMENDATIONS)
          .zipWithIndex
          .map { case ((course, score), index) => (course, score, index + 1) }

        // 添加到用户推荐列表
        if (recommendations.nonEmpty) {
          userRecommendations += ((userId, recommendations))
        } else {
          usersWithNoRecommendations += 1
        }
      }
    }

    // 打印统计信息
    println(s"\n推荐统计: 共为 ${userRecommendations.size} 名用户生成了推荐")
    println(s"有 $usersWithAllCourses 名用户已参加所有课程，使用了替代推荐策略")
    println(s"有 $usersWithNoRecommendations 名用户无法生成推荐")

    userRecommendations.toList
  }

  /**
   * 打印用户-课程矩阵(部分数据)
   */
  def printUserCourseMatrix(matrix: Map[String, Map[String, Double]]): Unit = {
    // 获取所有课程名列表(前5个)
    val allCourses = matrix.values.flatMap(_.keys).toSet.toSeq.sorted.take(5)
    val userSample = matrix.keys.toSeq.sorted.take(3)

    // 打印表头
    println("用户ID\t" + allCourses.mkString("\t"))

    // 打印每个用户的评分
    for (userId <- userSample) {
      val userRatings = matrix.getOrElse(userId, Map())
      val ratingsStr = allCourses.map(course =>
        f"${userRatings.getOrElse(course, 0.0)}%.1f"
      ).mkString("\t")

      println(s"$userId\t$ratingsStr")
    }
    println("... (更多数据省略)")
  }

  /**
   * 打印课程-课程相似度矩阵(部分数据)
   */
  def printCourseSimilarityMatrix(matrix: Map[String, Map[String, Double]]): Unit = {
    // 获取部分课程名作为示例
    val courseSample = matrix.keys.toSeq.sorted.take(4)

    // 打印表头
    println("课程\t" + courseSample.mkString("\t"))

    // 打印每个课程的相似度
    for (course1 <- courseSample) {
      val similarities = matrix.getOrElse(course1, Map())
      val similaritiesStr = courseSample.map(course2 =>
        if (course1 == course2) "1.00"
        else f"${similarities.getOrElse(course2, 0.0)}%.2f"
      ).mkString("\t")

      println(s"$course1\t$similaritiesStr")
    }
    println("... (更多数据省略)")
  }

  /**
   * 打印推荐结果
   */
  def printRecommendations(recommendations: List[(String, List[(String, Double, Int)])]): Unit = {
    println("+---------------+---------------------------------+----------+-------+")
    println("| 用户ID        | 推荐课程                         | 推荐得分  | 排名  |")
    println("+---------------+---------------------------------+----------+-------+")

    for ((userId, courseRecs) <- recommendations) {
      if (courseRecs.isEmpty) {
        println(s"| $userId | 无推荐课程 | --- | --- |")
      } else {
        val firstRec = courseRecs.head
        println(f"| ${userId}%-13s | ${firstRec._1}%-31s | ${firstRec._2}%8.2f | ${firstRec._3}%5d |")

        for (rec <- courseRecs.tail) {
          println(f"| ${""}%-13s | ${rec._1}%-31s | ${rec._2}%8.2f | ${rec._3}%5d |")
        }
      }
      println("+---------------+---------------------------------+----------+-------+")
    }
  }

  /**
   * 保存推荐结果到MySQL
   */
  def saveRecommendationsToDb(recommendations: List[(String, List[(String, Double, Int)])]): Unit = {
    var connection: Connection = null
    var statement: PreparedStatement = null

    try {
      connection = createConnection()
      statement = connection.prepareStatement(
        """
          |INSERT INTO user_recommendations
          |(student_id, course_name, recommendation_score, recommendation_type, rank_order, last_updated)
          |VALUES (?, ?, ?, 'CF', ?, NOW())
          |ON DUPLICATE KEY UPDATE
          |recommendation_score = VALUES(recommendation_score),
          |rank_order = VALUES(rank_order),
          |last_updated = NOW()
        """.stripMargin)

      for ((userId, courseRecs) <- recommendations) {
        for ((courseName, score, rank) <- courseRecs) {
          statement.setString(1, userId)
          statement.setString(2, courseName.replace(" (推荐复习)", "")) // 去掉标记文本保存到数据库
          statement.setDouble(3, score)
          statement.setInt(4, rank)
          statement.addBatch()
        }
      }

      statement.executeBatch()
    } catch {
      case e: Exception =>
        println(s"保存推荐结果时出错: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      if (statement != null) statement.close()
      if (connection != null) connection.close()
    }
  }

  /**
   * 保存课程相似度矩阵到MySQL
   */
  def saveCourseSimilarityToDb(similarityMatrix: Map[String, Map[String, Double]]): Unit = {
    var connection: Connection = null
    var statement: PreparedStatement = null

    try {
      connection = createConnection()
      statement = connection.prepareStatement(
        """
          |INSERT INTO course_similarity
          |(course1, course2, similarity_score, last_updated)
          |VALUES (?, ?, ?, NOW())
          |ON DUPLICATE KEY UPDATE
          |similarity_score = VALUES(similarity_score),
          |last_updated = NOW()
        """.stripMargin)

      for ((course1, similarities) <- similarityMatrix) {
        for ((course2, score) <- similarities) {
          statement.setString(1, course1)
          statement.setString(2, course2)
          statement.setDouble(3, score)
          statement.addBatch()
        }
      }

      statement.executeBatch()
    } catch {
      case e: Exception =>
        println(s"保存课程相似度矩阵时出错: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      if (statement != null) statement.close()
      if (connection != null) connection.close()
    }
  }

  /**
   * 保存用户课程矩阵到MySQL
   */
  def saveUserCourseMatrixToDb(userCourseMatrix: Map[String, Map[String, Double]]): Unit = {
    var connection: Connection = null
    var statement: PreparedStatement = null

    try {
      connection = createConnection()
      statement = connection.prepareStatement(
        """
          |INSERT INTO user_course_matrix
          |(student_id, course_name, attendance_rate, last_updated)
          |VALUES (?, ?, ?, NOW())
          |ON DUPLICATE KEY UPDATE
          |attendance_rate = VALUES(attendance_rate),
          |last_updated = NOW()
        """.stripMargin)

      for ((studentId, courses) <- userCourseMatrix) {
        for ((courseName, rating) <- courses) {
          // 将1-5的评分转换回百分比格式的出勤率 (0-100%)
          val attendanceRate = (rating - 1) * 25 // 转回0-100区间

          statement.setString(1, studentId)
          statement.setString(2, courseName)
          statement.setDouble(3, attendanceRate)
          statement.addBatch()
        }
      }

      statement.executeBatch()
    } catch {
      case e: Exception =>
        println(s"保存用户课程矩阵时出错: ${e.getMessage}")
        e.printStackTrace()
    } finally {
      if (statement != null) statement.close()
      if (connection != null) connection.close()
    }
  }

  /**
   * 创建MySQL数据库连接
   */
  def createConnection(): Connection = {
    // 加载数据库驱动
    Class.forName("com.mysql.cj.jdbc.Driver")

    // 创建连接 - 更新数据库名为attendance_new
    DriverManager.getConnection(
      "jdbc:mysql://43.143.125.94:3306/attendance_new?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai",
      "root",
      "Lwj378$$"
    )
  }
}